package com.mai.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mai.realtime.bean.TradeOrderBean;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

/**
 * @Creator: LBG
 * @CreateTime: 2022-09-01  09:35
 */
public class dws_trade_total {

    public static void main(String[] args) {
        System.setProperty("HADOOP_USER_NAME", "root");
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3013);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(5);

        env.enableCheckpointing(3000);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://39.108.71.119:8020/gmall/" + "dwd_trade_cart_add_inc");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(20 * 1000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        //从kafka读取数据
        Properties props=new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092");
        props.setProperty("isolation.level", "read_committed");
        props.setProperty("group.id", "dws_trade_total");
        props.setProperty("auto.offset.reset", "earliest");            //earliest  latest
        DataStreamSource<String> sourceStream = env.addSource(new FlinkKafkaConsumer<String>(
                "dwd_trade_order_detail_inc",
                new SimpleStringSchema(),
                props
        ));



        //watermark keyBy  window   reduce
        SingleOutputStreamOperator<TradeOrderBean> result = sourceStream
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("dt") * 1000;
                                    }
                                })
                )

                .keyBy(jo -> jo.getString("user_id"))
                //对每个元素进行处理
                .process(new KeyedProcessFunction<String, JSONObject, TradeOrderBean>() {
                    private ValueState<String> lastState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<TradeOrderBean> out) throws Exception {

                        String lastS = lastState.value();
                        String today = new SimpleDateFormat("yyyy-MM-dd").format(value.getLong("dt"));

                        Long single = 0L;

                        if (!today.equals(lastS)) {
                            single = 1L;
                            lastState.update(today);
                        }

                        out.collect(new TradeOrderBean("", "", "", single, 1L, value.getDouble("final_amount")));

                    }
                })
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TradeOrderBean>() {
                    @Override
                    public TradeOrderBean reduce(TradeOrderBean value1, TradeOrderBean value2) throws Exception {
                        value1.setOrderUniqueUserCount(value1.getOrderUniqueUserCount() + value2.getOrderUniqueUserCount());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setTotalMoney(value1.getTotalMoney() + value2.getTotalMoney());
                        return value1;
                    }
                }, new ProcessAllWindowFunction<TradeOrderBean, TradeOrderBean, TimeWindow>() {
                    @Override
                    public void process(Context ctx, Iterable<TradeOrderBean> elements, Collector<TradeOrderBean> out) throws Exception {
                        TradeOrderBean bean = elements.iterator().next();
                        String start = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(ctx.window().getStart());
                        String end = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(ctx.window().getEnd());
                        String curr = new SimpleDateFormat("yyyy-MM-dd").format(System.currentTimeMillis());

                        bean.setStt(start);
                        bean.setEdt(end);
                        bean.setCurDate(curr);

                        out.collect(bean);
                    }
                });



/*

CREATE TABLE dws_trade_total(
stt                     string,
    edt                     string,
    curDate                string,
    orderUniqueUserCount string,
    orderCount    string,
    totalMoney    string
    )
ENGINE = TinyLog();


         */




        //写入clickhouse    clickhouse --port 3579客户端
        result.addSink(new RichSinkFunction<TradeOrderBean>() {
            private Connection conn;
            private PreparedStatement insertStmt;
            @Override
            public void open(Configuration parameters) throws Exception {
                conn = DriverManager.getConnection(
                        "jdbc:clickhouse://hadoop162:8123/default"
                );
                insertStmt = conn.prepareStatement("INSERT INTO dws_trade_total (stt,edt,curDate,orderUniqueUserCount," +
                        "orderCount,totalMoney) VALUES (?, ?,?,?,?,?)");
            }

            // 来一条数据触发调用一次
            @Override
            public void invoke(TradeOrderBean value, Context ctx) throws Exception {
                insertStmt.setString(1, value.getStt());
                insertStmt.setString(2, value.getEdt());
                insertStmt.setString(3, value.getCurDate());
                insertStmt.setString(4, String.valueOf(value.getOrderUniqueUserCount()));
                insertStmt.setString(5, String.valueOf(value.getOrderCount()));
                insertStmt.setString(6, String.valueOf(value.getTotalMoney()));
                insertStmt.execute();

            }

            @Override
            public void close() throws Exception {

                insertStmt.close();
                conn.close();
            }
        });







        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
